package com.shujia.flink.sink

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema

//import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer.Semantic
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer, FlinkKafkaProducer}

object Demo3SinkKafka {

  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 读取kafka
      *
      */

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    properties.setProperty("group.id", "test")

    val myConsumer = new FlinkKafkaConsumer[String](
      "test",
      new SimpleStringSchema(),
      properties)


    val kafkaDS: DataStream[String] = env.addSource(myConsumer)


    val filterDS: DataStream[String] = kafkaDS.filter(word => !"java".equals(word))


    /**
      * 将数据保存到kafaz中
      *
      */

    val properties1 = new Properties
    properties1.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")


    val myProducer = new FlinkKafkaProducer[String](
      "filter-topic",
      new SimpleStringSchema(),
      properties1)


    filterDS.addSink(myProducer)


    env.execute()

  }

}
